package com.rem.flink.flink7Stream;

import com.rem.flink.flink2Source.ClickSource;
import com.rem.flink.flink2Source.Event;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 对流的操作
 * 分流 根据name 分流
 *
 * @author Rem
 * @date 2022-10-17
 */

public class SplitStreamByFilter {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource());

        SingleOutputStreamOperator<Event> maryStream = stream.filter(a -> "Mary".equals(a.getUser()));
        SingleOutputStreamOperator<Event> aliceStream = stream.filter(a -> "Alice".equals(a.getUser()));

        SingleOutputStreamOperator<Event> otherStream = stream.filter(a -> {
            String user = a.getUser();
            return !"Mary".equals(user) && !"Alice".equals(user);
        });

        maryStream.print("Mary pv");
        aliceStream.print("Alice pv");
        otherStream.print("other pv");

        env.execute();
    }
}
